package batch.controller;

import java.io.IOException;
import java.net.InetAddress;
import java.net.MalformedURLException;
import java.net.UnknownHostException;
import java.rmi.Naming;
import java.rmi.Remote;
import java.rmi.RemoteException;
import java.rmi.registry.LocateRegistry;
import java.rmi.registry.Registry;
import java.rmi.server.UnicastRemoteObject;
import java.sql.Connection;
import java.sql.SQLException;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.Date;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.Objects;
import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.ThreadFactory;
import java.util.concurrent.TimeUnit;

import javax.rmi.ssl.SslRMIClientSocketFactory;
import javax.rmi.ssl.SslRMIServerSocketFactory;

import org.apache.logging.log4j.LogManager;
import org.apache.logging.log4j.Logger;

import batch.status.Job;
import batch.status.JobDetail;
import batch.status.JobDetailStatus;
import batch.status.JobMaster;
import batch.status.JobMasterInfo;
import batch.status.JobState;
import batch.status.JobStatus;

import common.db.jdbc.Jdbc;

import core.config.Env;
import core.config.Factory;
import core.exception.PhysicalException;
import core.exception.ThreadExceptionHandler;
import core.exception.ThrowableUtil;
import core.util.DateUtil;

/**
 * バッチジョブ管理クラス
 *
 * @author Tadashi Nakayama
 * @version 1.0.0
 */
public final class JobManagerImpl extends UnicastRemoteObject implements JobManager, Runnable {
	/** serialVersionUID */
	private static final long serialVersionUID = 4825560113231843758L;

	/** ログ出力用クラス */
	private static final Logger LOGGER = LogManager.getLogger(JobManagerImpl.class);

	/** ジョブ拡張子 */
	private static final String ENV_BATCH_SUFFIX = "Batch.Suffix";
	/** バッチ起動最大数 */
	private static final String ENV_BATCH_MAXPROCS = "Batch.MaxProcs";
	/** バッチ起動間隔 */
	private static final String ENV_BATCH_INTERVAL = "Batch.Interval";
	/** 中止コマンド */
	private static final String ENV_BATCH_KILL = "Batch.KillCommand";

	/** 強制終了フラグ実現値 */
	private static final String ID_FORCE_KILL = "1";

	/** ジョブ管理マスタ */
	private final Map<String, JobMaster> jobMasterMap = new HashMap<>();
	/** デフォルト最大実行ジョブ数 */
	private final int maxProc;

	/**
	 * コンストラクタ
	 *
	 * @throws RemoteException リモート例外
	 */
	public JobManagerImpl() throws RemoteException {
		// 最大実行ジョブ数取得
		this.maxProc = Env.getEnv(ENV_BATCH_MAXPROCS, 20);
		LOGGER.info("MaxProc={}", String.valueOf(this.maxProc));
	}

	/**
	 * @see batch.controller.JobManager#cancel(long)
	 */
	@Override
	public void cancel(final long jobseq) throws RemoteException {
		JobStatus js = Factory.create(JobStatus.class);

		try (Connection conn = JobUtil.getConnection()) {
			conn.clearWarnings();
			conn.setReadOnly(false);
			conn.setAutoCommit(false);

			// ジョブ管理排他読込
			Job job = js.getJobWithLock(conn, jobseq);
			if (job == null) {
				return;
			}

			JobState sts = JobState.cancel(job.getJobSts());
			if (sts != null) {
				cancelJob(job.getJobId(), jobseq, conn);

				// ジョブ管理更新
				js.updateJobStatus(conn, jobseq, job.getHostId(), "", sts);

				conn.commit();
			}

		} catch (SQLException ex) {
			ThrowableUtil.error(ex);
			throw new RemoteException(ex.getMessage(), ex);
		}
	}

	/**
	 * ジョブキャンセル
	 *
	 * @param jobid ジョブID
	 * @param jobseq ジョブ連番
	 * @param conn コネクション
	 */
	private void cancelJob(final String jobid, final long jobseq, final Connection conn) {
		// 終了タイプ取得
		String type = getJobKillType(jobid);

		LOGGER.debug("type={}", type);

		if (!ID_FORCE_KILL.equals(type)) {
			return;
		}

		// ジョブ詳細管理排他読込
		JobDetailStatus jds = Factory.create(JobDetailStatus.class);
		List<JobDetail> l = jds.selectJobDetails(conn, jobseq);

		if (LOGGER.isDebugEnabled()) {
			if (l == null) {
				LOGGER.debug("l is null.");
			} else if (l.isEmpty()) {
				LOGGER.debug("l is zero.");
			} else {
				LOGGER.debug("job detail status={}", String.valueOf(l.get(0).getBatSts()));
			}
		}

		if (l == null || l.isEmpty() || !JobState.ID_B_EDIT.equals(l.get(0).getBatSts())) {
			return;
		}

		// 中止コマンド作成
		String cmd = Env.getEnv(ENV_BATCH_KILL);

		LOGGER.debug("job detail status={} command={}", String.valueOf(l.get(0).getBatSts()), cmd);

		if (!Objects.toString(cmd, "").trim().isEmpty()) {
			// ジョブ中止処理
			try {
				new ProcessBuilder(cmd, String.valueOf(l.get(0).getPid())).start();
			} catch (IOException ex) {
				LogManager.getLogger().warn(ex.getMessage(), ex);
			} finally {
				LOGGER.debug("executed={}", cmd);
			}
		}
	}

	/**
	 * プロセス起動管理処理
	 *
	 */
	@Override
	public void run() {
		LOGGER.info("JobManager Thread Start.");

		// ポーリング処理
		try {
			InetAddress localhost = InetAddress.getLocalHost();
			launch(localhost.getHostName());
		} catch (UnknownHostException ex) {
			LogManager.getLogger().error(ex.getMessage(), ex);
			throw new PhysicalException(ex);
		}

		LOGGER.info("JobManager Thread End.");
	}

	/**
	 * ジョブ管理マスタ情報取得
	 */
	private void getJobMaster() {
		try (Jdbc conn = JobUtil.getConnection()) {
			// ジョブ管理マスタ取得
			JobMasterInfo ji = Factory.create(JobMasterInfo.class);
			List<JobMaster> l = ji.getJobMasterInfoAll(conn);
			if (l != null) {
				for (final JobMaster obj : l) {
					this.jobMasterMap.put(obj.getJobId(), obj);
				}
			}
		}
	}

	/**
	 * 起動処理
	 *
	 * @param host ホスト名
	 * @return 起動数
	 */
	private int launch(final String host) {

		int launchedProc = 0;

		try (Connection conn = JobUtil.getConnection()) {
			// ジョブステータス取得
			JobStatus js = Factory.create(JobStatus.class);
			List<Job> active = js.getJobListByStatus(conn, host, JobState.running());
			// 実行プロセス数チェック
			if (this.maxProc <= active.size()) {
				return launchedProc;
			}

			// 処理依頼中のプロセスを登録日時順に取得
			List<Job> queue = js.getJobListByStatus(conn, null, JobState.requested());

			// 最大実行ジョブ数になるか、処理依頼中がなくなるまでプロセス起動
			LOGGER.debug("RequestCount={}", String.valueOf(queue.size()));

			// 現在時刻取得
			String now = DateUtil.format(new Date(), "HHmm");

			for (final Job ap : queue) {
				String jid = ap.getJobId().trim();

				// 中止依頼
				// 起動可能時刻チェック
				if (!checkLaunchTime(jid, now)) {
					continue;
				}

				// ジョブの実行多重度チェック
				if (getMultipleCount(jid) <= getActiveProcCount(active, jid)) {
					continue;
				}

				// バッチ起動シェルの実行
				if (execBatchShell(ap.getJobSeq(), jid, host, conn)) {
					launchedProc++;
					active.add(ap);
				}

				conn.commit();
			}

		} catch (SQLException ex) {
			ThrowableUtil.error(ex);
		}

		return launchedProc;
	}

	/**
	 * ジョブ中止方法取得
	 *
	 * @param jid ジョブID
	 * @return 中止方法
	 */
	private String getJobKillType(final String jid) {
		JobMaster rec = this.jobMasterMap.get(jid);
		if (rec != null) {
			return rec.getCancelType();
		}
		return "";
	}

	/**
	 * ジョブ起動時間チェック
	 *
	 * @param jid ジョブID
	 * @param now 現在時刻
	 * @return 起動可否
	 */
	private boolean checkLaunchTime(final String jid, final String now) {
		JobMaster rec = this.jobMasterMap.get(jid);
		return rec == null
				|| JobUtil.checkInTime(now, rec.getRunnableBegin(), rec.getRunnableEnd());
	}

	/**
	 * 処理中プロセス数取得
	 *
	 * @param list リスト
	 * @param jid ジョブID
	 * @return 処理依頼中プロセスVector
	 */
	private int getActiveProcCount(final List<Job> list, final String jid) {
		int ret = 0;
		if (list != null) {
			for (final Job vals : list) {
				if (jid.equals(vals.getJobId())) {
					ret++;
				}
			}
		}
		return ret;
	}

	/**
	 * プロセス多重度取得
	 *
	 * @param jid ジョブID
	 * @return 多重度
	 */
	private int getMultipleCount(final String jid) {
		JobMaster rec = this.jobMasterMap.get(jid);
		if (rec != null) {
			return rec.getMultiplicity();
		}
		return 1;
	}

	/**
	 * バッチ起動シェル実行
	 *
	 * @param jobseq ジョブ連番
	 * @param jid ジョブID
	 * @param host ホスト名
	 * @param conn コネクション
	 * @return 実行結果
	 */
	private boolean execBatchShell(final long jobseq, final String jid,
			final String host, final Connection conn) {
		// レコードにロックをかける
		JobStatus js = Factory.create(JobStatus.class);
		Job job = js.getJobWithLock(conn, jobseq);
		if (job == null || !JobState.isRequest(job.getJobSts())) {
			return false;
		}

		JobState sts = JobState.ID_B_SH_START;
		try {
			new ProcessBuilder(newCommands(jobseq, jid, job)).start();
		} catch (IOException ex) {
			LogManager.getLogger().error(ex.getMessage(), ex);
			sts = JobState.ID_B_INVALID;
		}

		// バッチ管理更新
		js.updateJobStatus(conn, jobseq, host, "", sts);

		return JobState.ID_B_SH_START.equals(sts);
	}

	/**
	 * コマンドリスト作成
	 *
	 * @param jobseq ジョブ連番
	 * @param jid ジョブID
	 * @param job ジョブ管理
	 * @return コマンドリスト
	 */
	private List<String> newCommands(final long jobseq, final String jid, final Job job) {
		// バッチ起動シェルコマンド作成
		String suffix = Env.getEnv(ENV_BATCH_SUFFIX);

		LOGGER.debug("{}{} {} {} ", jid, suffix, ID_B_ONLINE_OPTION, String.valueOf(jobseq));

		// バッチ起動処理
		List<String> cmd = new ArrayList<>();
		cmd.add(jid + suffix);
		cmd.add(ID_B_ONLINE_OPTION);
		cmd.add(String.valueOf(jobseq));
		cmd.addAll(Arrays.asList(job.getJobParam()));
		return cmd;
	}

	/**
	 * クリーン指定
	 * @param args 引数
	 */
	private static void clean(final String... args) {
		for (final String arg : args) {
			if ("-clean".equals(arg)) {
				// TODO DB整合処理

				return;
			}
		}
	}

	/**
	 * JobManager取得
	 */
	private void start() {
		// スレッド起動
		ScheduledExecutorService ses =
				Executors.newSingleThreadScheduledExecutor(new JobThreadFactory());
		ses.scheduleWithFixedDelay(this, TimeUnit.SECONDS.toMillis(30),
				Env.getEnv(ENV_BATCH_INTERVAL, 10000), TimeUnit.MILLISECONDS);
	}

	/**
	 * メイン処理
	 *
	 * @param args 起動パラメタ
	 */
	public static void main(final String... args) {
		// セキュリティマネージャ設定
		if (System.getSecurityManager() == null) {
			System.setSecurityManager(new SecurityManager());
		}

		clean(args);

		// バインディング
		try {
			// JobManager取得
			JobManagerImpl jm = new JobManagerImpl();
			jm.getJobMaster();
			jm.start();

			int ssl = JobRequestor.getSslPort();
			if (0 < ssl) {
				Remote obj = UnicastRemoteObject.exportObject(jm, ssl,
						new SslRMIClientSocketFactory(),
						new SslRMIServerSocketFactory());

				Registry registry = LocateRegistry.createRegistry(ssl,
						new SslRMIClientSocketFactory(),
						new SslRMIServerSocketFactory());
				registry.rebind("//localhost" + BATCH_BIND_NAME, obj);
			} else {
				Naming.rebind(JobRequestor.toBatchUri("//localhost", BATCH_BIND_NAME), jm);
			}

			LOGGER.debug("End Bind");

		} catch (RemoteException | MalformedURLException ex) {
			ThrowableUtil.error(ex);
			Runtime.getRuntime().exit(1);
		}
	}

	/**
	 * JobThreadFactory
	 * @author Tadashi Nakayama
	 */
	private static final class JobThreadFactory implements ThreadFactory {
		/**
		 * コンストラクタ
		 */
		public JobThreadFactory() {
			super();
		}

		/**
		 * @see java.util.concurrent.ThreadFactory#newThread(java.lang.Runnable)
		 */
		@Override
		public Thread newThread(final Runnable r) {
			Thread th = new Thread(r, "JobManagerImpl");
			th.setUncaughtExceptionHandler(new ThreadExceptionHandler());
			return th;
		}
	}
}
